package fenxiTuijian

import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object tuijian {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("test")
      .enableHiveSupport()
      .getOrCreate()

    // todo 首先读取数据然后下面对数据进行预处理
    var order = spark.table("work.order_clean")
    var median = spark.table("work.media_clean")

    median.createTempView("median")


    //  todo 统计每个用户其观看频道的次数
    val r1 = spark.sql(
      """
        |select distinct
        |phone_no,
        |station_name,
        |count(*) over(partition by phone_no,station_name) as count
        |from median
        |""".stripMargin)

    r1.orderBy(desc("count")).show


    //  todo 根据观看次数构建评分列
    val r2 = r1.withColumn(
      "rating",
      when(col("count") <= 30, 1)
        .when(col("count") > 30 and col("count") <= 40, 2)
        .when(col("count") > 40 and col("count") <= 60, 3)
        .when(col("count") > 60 and col("count") <= 100, 40)
        .when(col("count") > 100, 5)
    )

    r2.orderBy(desc("count")).show(500)


    //  todo 使用StringIndexer将频道转化为数值类型
    val indexer:StringIndexer= new StringIndexer()
      .setInputCol("station_name")
      .setOutputCol("station_id")

    //  todo 对频道字段进行转换
    val index_data=indexer.fit(r2).transform(r2)

    //  todo 对用户生成唯一的id
    val userIndexer = new StringIndexer()
      .setInputCol("phone_no")
      .setOutputCol("user_id") // 输出列名

    //  todo 对用户字段进行转换
    val data = userIndexer.fit(index_data).transform(index_data)

    //  todo 单独取出频道和用户原本的id
    val user_station=data.select("phone_no","user_id","station_name","station_id")
      .withColumnRenamed("user_id","userid")
      .withColumnRenamed("station_id","stationid")

    user_station.show


    //  todo 构建模型进行推荐
    val als = new ALS()
      .setUserCol("user_id")
      .setItemCol("station_id")
      .setRatingCol("rating")
      .setColdStartStrategy("drop")     //  处理冷启动问题

    //  todo 构建模型
    val model=als.fit(data)

    //  todo 对每个用户推荐5个频道
    val to_user_result=model.recommendForAllUsers(5)
    //  todo 对每个频道推荐5个用户
    val to_station_result=model.recommendForAllItems(5)

//    to_user_result.columns.foreach(println)

    println("对每个用户推荐5个频道")
    to_user_result.join(user_station,user_station("userid")===to_user_result("user_id"),"left")
      .select("userid", "phone_no", "recommendations")
      .distinct()
      .show(false)


    println("对每个频道推荐5个用户")
    to_station_result.join(user_station,user_station("stationid")===to_station_result("station_id"),"left")
      .select("station_id","station_name","recommendations")
      .distinct()
      .show(false)





  }

}
